package com.atguigu.app.dwd;

import com.atguigu.utils.MyKafkaUtil;
import com.atguigu.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @className: DwdTradeCartAdd
 * @author: LinCong
 * @description:    维度退化 消费topic_db中gmall数据库cart_info表的数据，
 *                  将其和mysql gmall数据库维度表base_dic中对应主键的数据关联，
 *                  然后做维度退化，最后输出到kafka
 * @date: 2023/1/31 14:43
 * @version: 1.0
 */

//业务服务器（mysql）-> maxwell -> kafka(ods) -> flink(DwdTradeCartAdd) -> kafka
public class DwdTradeCartAdd {
    public static void main(String[] args) throws Exception {
//        1、获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        ////        1.1、开启checkpoint
//        env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
//        //设置checkpoint的超时时间,如果 Checkpoint在 10分钟内尚未完成说明该次Checkpoint失败,则丢弃。(默认10分钟)
//        env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(120000L);
//        //固定延迟重启   （最多重启次数，每次重启的时间间隔）
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));
////        1.2、设置状态后端
//        env.setStateBackend(new HashMapStateBackend());
//        System.setProperty("HADOOP_USER_NAME", "kevin");
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop3cluster/211126/ck");

//        2、使用ddl读取topic_db主题的数据创建表
        tableEnv.executeSql(MyKafkaUtil.getTopicDb("Cart_Add"));

//        3、过滤出加购数据
        Table cartAddTable = tableEnv.sqlQuery("select   " +
                "    `data`['id'] id,   " +
                "    `data`['user_id'] user_id,   " +
                "    `data`['sku_id'] sku_id,   " +
                "    `data`['cart_price'] cart_price,   " +
                "    if(`type`='insert',`data`['sku_num'],cast(cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int) as string)) sku_num,   " +
                "    `data`['sku_name'] sku_name,   " +
                "    `data`['is_checked'] is_checked,   " +
                "    `data`['create_time'] create_time,   " +
                "    `data`['operate_time'] operate_time,   " +
                "    `data`['is_ordered'] is_ordered,   " +
                "    `data`['order_time'] order_time,   " +
                "    `data`['source_type'] source_type,   " +
                "    `data`['source_id'] source_id,   " +
                "    pt   " +
                "from topic_db   " +
                "where `database` = 'gmall'   " +
                "and `table` = 'cart_info'   " +
                "and (`type` = 'insert'   " +
                "or (`type` = 'update'    " +
                "    and    " +
                "    `old`['sku_num'] is not null    " +
                "    and    " +
                "    cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int)))");

//        将加购表转化为流再打印
//        tableEnv.toAppendStream(cartAddTable, Row.class)
//                .print(">>>>>>");
        tableEnv.createTemporaryView("cart_info_table", cartAddTable);

//        4、读取mysql的base_dic表作为LookUp表
        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());

//        5、关联两张表
        Table cartAddWithDicTable = tableEnv.sqlQuery("select   " +
                "    ci.id,   " +
                "    ci.user_id,   " +
                "    ci.sku_id,   " +
                "    ci.cart_price,   " +
                "    ci.sku_num,   " +
                "    ci.sku_name,   " +
                "    ci.is_checked,   " +
                "    ci.create_time,   " +
                "    ci.operate_time,   " +
                "    ci.is_ordered,   " +
                "    ci.order_time,   " +
                "    ci.source_type source_type_id,   " +
                "    dic.dic_name source_type_name,   " +
                "    ci.source_id   " +
                "from cart_info_table ci   " +
                "join base_dic FOR SYSTEM_TIME AS OF ci.pt AS dic   " +
                "on ci.source_type=dic.dic_code");
        tableEnv.createTemporaryView("cart_add_dic_table",cartAddWithDicTable);

//        6、使用ddl方式创建加购事实表
        tableEnv.executeSql("create table dwd_cart_add(   " +
                "    `id` STRING,   " +
                "    `user_id` STRING,   " +
                "    `sku_id` STRING,   " +
                "    `cart_price` STRING,   " +
                "    `sku_num` STRING,   " +
                "    `sku_name` STRING,   " +
                "    `is_checked` STRING,   " +
                "    `create_time` STRING,   " +
                "    `operate_time` STRING,   " +
                "    `is_ordered` STRING,   " +
                "    `order_time` STRING,   " +
                "    `source_type_id` STRING,   " +
                "    `source_type_name` STRING,   " +
                "    `source_id` STRING   " +
                ")"+MyKafkaUtil.getKafkaSinkDDL("dwd_trade_cart_add"));

//        7、将数据写出
            tableEnv.executeSql("insert into dwd_cart_add select * from cart_add_dic_table")
            .print();

//        8、启动任务
        env.execute("DwdTradeCartAdd");
    }
}
